package com.atguigu.day08;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.UrlViewCountPerWindow;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

// 触发器
// 每隔10秒钟才能看到一个窗口的统计信息，太慢了
// 也就是说等窗口闭合的时候，才能看到窗口的统计结果
// 能不能在窗口没有闭合的时候，就查看窗口的统计结果呢？
// 窗口大小是10秒钟，但想在每个整数秒看一次窗口的计算结果
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .filter(r -> r.url.equals("./home"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                return element.ts;
                            }
                        })
                )
                .keyBy(r -> r.url)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .trigger(new MyTrigger())
                .aggregate(
                        new AggregateFunction<ClickEvent, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(ClickEvent value, Long accumulator) {
                                return accumulator + 1L;
                            }

                            @Override
                            public Long getResult(Long accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return null;
                            }
                        },
                        new ProcessWindowFunction<Long, UrlViewCountPerWindow, String, TimeWindow>() {
                            @Override
                            public void process(String s, Context context, Iterable<Long> elements, Collector<UrlViewCountPerWindow> out) throws Exception {
                                out.collect(new UrlViewCountPerWindow(
                                        s,
                                        elements.iterator().next(),
                                        context.window().getStart(),
                                        context.window().getEnd()
                                ));
                            }
                        }
                )
                .print();

        env.execute();
    }

    // 当窗口中的第一条数据到达窗口时，第一条数据的时间戳之后的整数秒，输出窗口统计结果
    // 例如第一条数据的时间戳是1234ms，在2s，3s，....，8s，9s时输出窗口统计结果
    public static class MyTrigger extends Trigger<ClickEvent, TimeWindow> {
        // 每来一条数据，调用一次
        @Override
        public TriggerResult onElement(ClickEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            // 只注册第一条数据的时间戳后面的整数秒的定时器
            // 窗口状态变量
            ValueState<Boolean> isFirstElement = ctx.getPartitionedState(
                    new ValueStateDescriptor<Boolean>(
                            "is-first",
                            Types.BOOLEAN
                    )
            );
            // 只有当第一条数据到达的时候，才会进入条件分支
            if (isFirstElement.value() == null) {
                // 说明窗口中的第一条数据到达
                // 计算element.ts紧挨着的整数秒
                // 1234 + 1000 - 1234 % 1000 = 2000
                long nextTs = element.ts + 1000L - element.ts % 1000L;
                // 注册的是onEventTime定时器
                // 假设第一条数据的时间戳是1234毫秒，那么注册一个2000ms的定时器
                ctx.registerEventTimeTimer(nextTs);
                // 将标志位置为true
                isFirstElement.update(true);
            }
            // 窗口不做任何操作
            return TriggerResult.CONTINUE;
        }

        // 当机器时间到达`time`时，触发调用
        // 处理时间定时器
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        // 水位线到达`time`时，触发调用
        // 事件时间定时器
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            long nextTs = time + 1000L;
            if (nextTs < window.getEnd()) {
                // 注意！！！这里注册的定时器还是onEventTime，也就是方法本身
                ctx.registerEventTimeTimer(nextTs);
            }
            // FIRE触发的是trigger后面的aggregate方法的执行
            return TriggerResult.FIRE;
        }

        // 当窗口闭合销毁时调用
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            // 触发器的窗口状态变量无法在全局定义，所以这里需要再次实例化，单例
            ValueState<Boolean> isFirstElement = ctx.getPartitionedState(
                    new ValueStateDescriptor<Boolean>(
                            "is-first",
                            Types.BOOLEAN
                    )
            );
            // 清空窗口状态变量
            isFirstElement.clear();
        }
    }
}
